package com.hengxinguotong.zhihuichengjian.net;

import android.content.Context;


import com.hengxinguotong.zhihuichengjian.bean.DownloadInfo;
import com.hengxinguotong.zhihuichengjian.database.DownloadDAO;
import com.hengxinguotong.zhihuichengjian.exception.ApiException;

import java.io.File;
import java.io.InputStream;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
import okhttp3.ResponseBody;
import retrofit2.Call;
import retrofit2.Response;
import retrofit2.Retrofit;

public class Downloader
{
    private static final String TAG = "Downloader";

    private DownloadDAO dao = null;// 数据库操作工具类

    private String urlstr = null;// 下载的地址
    private String localfile = null;// 保存路径
    private int threadcount = 0;// 线程数
    private List<DownloadInfo> infos = null;

    private OnDownload onDownload = null;//结果处理

    private static final int INIT = 1;// 定义三种下载的状态：初始化状态，正在下载状态，暂停状态
    private static final int DOWNLOADING = 2;
    private static final int PAUSE = 3;
    private int state = INIT;// 当前状态

    private DownloadService downloadService;

    public Downloader(Builder builder)
    {
        this.dao = builder.dao;
        this.urlstr = builder.urlstr;
        this.localfile = builder.localfile;
        this.threadcount = builder.threadcount;
        this.onDownload = builder.onDownload;
        this.downloadService = createDownloadService(urlstr);
    }

    /**
     * 创建下载服务对象
     *
     * @param url
     * @return
     */
    private DownloadService createDownloadService(String url)
    {
        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("http://eshenghuo.escst.com/app/proprietor/")
                .client(ClientFactory.create(url))
                .build();
        return retrofit.create(DownloadService.class);
    }

    /**
     * 下载开始，启动下载线程
     */
    public void download()
    {
        if (state == DOWNLOADING)
        {
            return;
        }
        Observable.create(new ObservableOnSubscribe<Object>()
        {
            @Override
            public void subscribe(ObservableEmitter<Object> e) throws Exception
            {
                boolean isFirst = dao.isHasInfors(urlstr);
                File file = new File(localfile);
                if (isFirst || !file.exists())
                {
                    dao.delete(urlstr);
                    file = new File(localfile);
                    if (!file.getParentFile().exists())
                    {
                        file.getParentFile().mkdirs();
                    }
                    if (!file.exists())
                    {
                        file.createNewFile();
                    }
                    Call<ResponseBody> call = downloadService.downloadFile(urlstr);
                    Response<ResponseBody> response = call.execute();
                    if (response.code() != 200)
                    {
                        throw new ApiException(response.message());
                    }
                    ResponseBody responseBody = response.body();
                    long fileSize = responseBody.contentLength();
                    long range = fileSize / threadcount;
                    infos = new ArrayList<>();
                    for (int i = 0; i < threadcount - 1; i++)
                    {
                        DownloadInfo info = new DownloadInfo(urlstr);
                        info.setThreadId(i);
                        info.setStartPos(i * range);
                        info.setEndPos((i + 1) * range - 1);
                        info.setCompeleteSize(0);
                        infos.add(info);
                    }
                    DownloadInfo info = new DownloadInfo(urlstr);
                    info.setThreadId(threadcount - 1);
                    info.setStartPos((threadcount - 1) * range);
                    info.setEndPos(fileSize - 1);
                    info.setCompeleteSize(0);
                    infos.add(info);
                    dao.saveInfos(infos);
                    responseBody.close();
                } else
                {
                    infos = dao.getInfos(urlstr);
                }
                e.onNext(infos);
                e.onComplete();
                for (int i = 0; i < infos.size(); i++)
                {
                    downloadThread(i);
                }
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Object>()
                {
                    @Override
                    public void onSubscribe(Disposable d)
                    {
                        state = DOWNLOADING;
                    }

                    @Override
                    public void onNext(Object o)
                    {
                        long fileComplete = 0;
                        long fileSize = 0;
                        for (DownloadInfo info : infos)
                        {
                            fileComplete += info.getCompeleteSize();
                            fileSize += info.getEndPos() - info.getStartPos() + 1;
                        }
                        if (onDownload != null)
                        {
                            onDownload.onStart(fileSize, fileComplete);
                        }
                    }

                    @Override
                    public void onError(Throwable e)
                    {
                        e.printStackTrace();
                        if (onDownload != null)
                        {
                            onDownload.onError();
                        }
                    }

                    @Override
                    public void onComplete()
                    {

                    }
                });
    }

    /**
     * 下载线程
     *
     * @param index
     */
    private void downloadThread(final int index)
    {
        Observable.create(new ObservableOnSubscribe<Long>()
        {
            @Override
            public void subscribe(ObservableEmitter<Long> e) throws Exception
            {
                DownloadInfo info = infos.get(index);
                int threadId = info.getThreadId();
                long startPos = info.getStartPos();
                long compeleteSize = info.getCompeleteSize();
                long endPos = info.getEndPos();
                StringBuilder builder = new StringBuilder("bytes=");
                builder.append(startPos + compeleteSize)
                        .append("-").append(endPos);
                Call<ResponseBody> call = downloadService.downloadFile(builder.toString(),
                        info.getUrl());
                Response<ResponseBody> response = call.execute();
                if (response.code() != 206)
                {
                    throw new ApiException(response.message());
                }
                ResponseBody responseBody = response.body();
                InputStream inputStream = responseBody.byteStream();
                RandomAccessFile randomAccessFile = new RandomAccessFile(localfile, "rwd");
                randomAccessFile.seek(startPos + compeleteSize);
                byte[] buffer = new byte[1024 * 6];
                int length = -1;
                while ((length = inputStream.read(buffer)) != -1)
                {
                    randomAccessFile.write(buffer, 0, length);
                    compeleteSize += length;
                    dao.updataInfos(threadId, compeleteSize, info.getUrl());
                    info.setCompeleteSize(compeleteSize);
                    e.onNext(compeleteSize);
                }
                inputStream.close();
                randomAccessFile.close();
                responseBody.close();
                if ((startPos + compeleteSize) != (endPos + 1))
                {
                    throw new ApiException("Data Error");
                }
                info.setThreadEnd(true);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new DownObserver(index));
    }

    /**
     * 下载进度和结果回调
     */
    private class DownObserver implements Observer<Long>
    {
        private DownloadInfo info;

        public DownObserver(int index)
        {
            this.info = infos.get(index);
        }

        @Override
        public void onSubscribe(Disposable d)
        {
            //TODO 可以把disposable收集起来，用于在下载被中断时，取消订阅，停止下载线程
        }

        @Override
        public void onNext(Long aLong)
        {
            if (onDownload != null)
            {
                long complete = 0;
                for (DownloadInfo info : infos)
                {
                    complete += info.getCompeleteSize();
                }
                onDownload.onProgress(complete);
            }
        }

        @Override
        public void onError(Throwable e)
        {
            e.printStackTrace();
            info.setThreadError(true);
            if (onDownload != null)
            {
                onDownload.onThreadError(info.getThreadId(), e.getMessage());
            }
            boolean allError = true;
            for (DownloadInfo info : infos)
            {
                if (!info.isThreadError())
                {
                    allError = false;
                    break;
                }
            }
            if (allError && onDownload != null)
            {
                onDownload.onError();
            }
        }

        @Override
        public void onComplete()
        {
            if (onDownload != null)
            {
                onDownload.onThreadEnd(info.getThreadId());
            }
            boolean allEnd = true;
            for (DownloadInfo info : infos)
            {
                if (!info.isThreadEnd())
                {
                    allEnd = false;
                    break;
                }
            }
            if (allEnd)
            {
                dao.delete(urlstr);
                if (onDownload != null)
                {
                    onDownload.onEnd();
                }
            }
        }
    }

    /**
     * 下载对象构造类
     */
    public static class Builder
    {
        private DownloadDAO dao;// 数据库操作工具类
        private String urlstr;// 下载的地址
        private String localfile;// 保存路径
        private int threadcount;// 线程数
        private OnDownload onDownload;

        public Builder(Context context)
        {
            dao = new DownloadDAO(context);
        }

        public Builder remoteUrl(String urlstr)
        {
            this.urlstr = urlstr;
            return this;
        }

        public Builder localUrl(String localfile)
        {
            this.localfile = localfile;
            return this;
        }

        public Builder threadCount(int threadcount)
        {
            this.threadcount = threadcount;
            return this;
        }

        public Builder callback(OnDownload onDownload)
        {
            this.onDownload = onDownload;
            return this;
        }

        public Downloader build()
        {
            return new Downloader(this);
        }
    }

    /**
     * 下载观察者
     */
    public interface OnDownload
    {
        void onStart(long fileSize, long complete);

        void onProgress(long complete);

        void onThreadEnd(int threadId);

        void onThreadError(int threadId, String error);

        void onEnd();

        void onError();
    }
}
